/**
 * Created by lzfto
 * Created time: 2025/05/25
 */
package org.lzfto.flink.demo.table.service.impl;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.lzfto.flink.demo.common.Configuration;
import org.lzfto.flink.demo.table.service.FlinkConnectorTestService;

import java.util.concurrent.ExecutionException;

/**
 * @author lzfto
 * @created 2025年6月2日 10:17:48
 */
public class CsvFileConnectorTest implements FlinkConnectorTestService {

	@Override
	public void run() {
		// TODO Auto-generated method stub
		// 创建执行环境
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tabEnv = StreamTableEnvironment.create(env); // 来自于bridge
		env.setParallelism(1);

		// 利用文件连接器(csv)创建一个输入表
		//file:///E:/learning/learn-flink/java/first-tableapi/student_score.csv
		String dir=Configuration.getConf().getEnvDirectory();
		String file="file:///"+dir+"student_score.csv";
		String ddlSql = """
					create table student_score(
					   name   String,
					   score  double
					) with (
					  'connector'='filesystem',
					  'path'='%s',
					  'format'='csv',
					  'csv.file-delimiter'=',',
					  'csv.ignore-parse-errors'='true'
					)
				""".formatted(file);
		tabEnv.executeSql(ddlSql);

		String exportDdlSql = """
				 create temporary table SinkTable(
				        name   String,
				        score  double
				   )
				   with('connector'='print')
				""";

		tabEnv.executeSql(exportDdlSql);

		Table importTable = tabEnv.from("student_score");
		TableResult result = importTable.executeInsert("SinkTable");
        try {
            result.await();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        } catch (ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

}
